package org.atguigu.gmall.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.db
 * Author: ZengHaiFeng
 * CreateTime:2023/4/26 16:54:57
 * Description: Todo
 * Version：1.0
 */
public class Dwd_08_DwdTradeRefundPaySuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_08_DwdTradeRefundPaySuc().init(30008,2,"Dwd_08_DwdTradeRefundPaySuc");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        // read data from ods_db
        readOdsDb(tbe,"Dwd_08_DwdTradeRefundPaySuc");
        // read data from base_dic
        readBaseDic(tbe);
        // filter refund pay success data
        Table refundPaySuc = tbe.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['order_id'] order_id, " +
                " `data`['sku_id'] sku_id, " +
                " `data`['payment_type'] payment_type, " +
                " `data`['callback_time'] callback_time, " +
                " `data`['total_amount'] total_amount, " +
                " `pt`," +
                " `ts`  " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='refund_payment' and " +
                " `type`='update' and " +
                " `old`['refund_status'] is not null and " +
                " `data`['refund_status']='1602' ");
        tbe.createTemporaryView("refund_pay_suc",refundPaySuc);

        // filter refund info data
        Table refundInfo = tbe.sqlQuery("select " +
                " `data`['order_id'] order_id, " +
                " `data`['sku_id'] sku_id, " +
                " `data`['refund_num'] refund_num, " +
                " `old` " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_refund_info' and " +
                " `type`='update' and " +
                " `old`['refund_status'] is not null and " +
                " `data`['refund_status']='0705' ");
        tbe.createTemporaryView("refund_info",refundInfo);

        // filter refund pay success order info data
        Table refundPaySucOrderInfo = tbe.sqlQuery(" select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['province_id'] province_id, " +
                " `old` " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_info' and " +
                " `type`='update' and " +
                " `old`['order_status'] is not null and " +
                " `data`['order_status']='1006' ");
        tbe.createTemporaryView("refund_pay_success_order_info",refundPaySucOrderInfo);

        // join
        Table result = tbe.sqlQuery(" select " +
                " rp.id, " +
                " rs.user_id, " +
                " rp.order_id, " +
                " rp.sku_id, " +
                " rs.province_id, " +
                " rp.payment_type payment_type_code," +
                " dic.dic_name payment_type_name, " +
                " date_format(rp.callback_time,'yyyy-MM-dd') date_id, " +
                " rp.callback_time, " +
                " ri.refund_num, " +
                " rp.total_amount, " +
                " rp.ts " +
                " from refund_pay_suc rp join refund_info ri " +
                " on rp.order_id=ri.order_id and rp.sku_id=ri.sku_id " +
                " join refund_pay_success_order_info rs " +
                " on rp.order_id=rs.id " +
                " join base_dic for system_time as of rp.pt as dic " +
                " on rp.payment_type=dic.dic_code ");

        // output
        tbe.executeSql("create table dwd_trade_refund_pay_suc(" +
                " id string, " +
                " user_id string, " +
                " order_id string, " +
                " sku_id string, " +
                " province_id string, " +
                " payment_type_code string, " +
                " payment_type_name string, " +
                " date_id string, " +
                " callback_time string, " +
                " refund_num string, " +
                " total_amount string, " +
                " ts bigint " +
                ") " + SQLUtil.getKafkaSinkDDL( Constant.TOPIC_DWD_TRADE_REFUND_PAY_SUC));
        result.executeInsert("dwd_trade_refund_pay_suc");
    }
}
